package net.flyinggroup.introduction;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;

import static net.flyinggroup.introduction.Constants.TCP_PORT;

@Slf4j
public class Server {
    private final Disposable server;
    private final DataPublisher publisher = new DataPublisher();
    private final GameController gameController;

    public Server() {
        server = RSocketServer.create((setup, sendingSocket) -> Mono.just(new RSocketImpl()))
                .bind(TcpServerTransport.create("localhost", TCP_PORT))
                .subscribe(closeableChannel -> log.info("onNext" + "........."),
                        throwable -> log.info("onError" + "........."),
                        () -> log.info("onComplete" + "........."),
                        subscription -> log.info("onSubscription" + "........."));

        gameController = new GameController("Server Player");
    }

    public void dispose() {
        publisher.complete();
        server.dispose();
    }

    private class RSocketImpl implements RSocket {
        /**
         * Request-Response interaction model of {@code RSocket}.
         *
         * @param payload Request payload.
         * @return {@code Publisher} containing at most a single {@code Payload} representing the
         * response.
         */
        @Override
        public Mono<Payload> requestResponse(Payload payload) {
            return Mono.just(payload);
        }

        /**
         * Fire and Forget interaction model of {@code RSocket}.
         *
         * @param payload Request payload.
         * @return {@code Publisher} that completes when the passed {@code payload} is successfully
         * handled, otherwise errors.
         */
        @Override
        public Mono<Void> fireAndForget(Payload payload) {
            publisher.publish(payload);
            log.info("received:" + payload.getData().getFloat());
            return Mono.empty();
        }

        /**
         * Request-Stream interaction model of {@code RSocket}.
         *
         * @param payload Request payload.
         * @return {@code Publisher} containing the stream of {@code Payload}s representing the response.
         */
        @Override
        public Flux<Payload> requestStream(Payload payload) {
            return Flux.from(publisher);
        }

        /**
         * Request-Channel interaction model of {@code RSocket}.
         *
         * @param payloads Stream of request payloads.
         * @return Stream of response payloads.
         */
        @Override
        public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
            Flux.from(payloads).subscribe(payload -> log.info("Channel request from client: {}", payload.getDataUtf8()));
            return Flux.interval(Duration.ofSeconds(2))
                    .map(l -> DefaultPayload.create(String.valueOf(l + 10000)))
                    .take(3);
        }
    }

    private class DataPublisher implements Publisher<Payload> {
        private Subscriber<? super Payload> subscriber;

        @Override
        public void subscribe(Subscriber<? super Payload> subscriber) {
            this.subscriber = subscriber;
        }

        public void publish(Payload payload) {
            if (subscriber != null) subscriber.onNext(payload);
        }

        public void complete() {
            if (subscriber != null) subscriber.onComplete();
        }
    }
}
